package org.demo.zookeeper;

import java.util.Map;
import java.util.WeakHashMap;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.RevocationListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
/**
 *  底层的机制非常的简单: "获取锁"的操作,就是在zookeeper中创建一个EPHEMERAL_SEQUENTIAL节点,
 *  同时对此节点的临近节点注册一个watcher;当"临近节点"被删除时,表示其他进程已经释放了锁,此watcher将会触发,并唤醒当前线程,
 *  然后acquire方法返回.."释放锁"的操作,就是删除此临时节点.此时临近的下一个节点将获得锁..

    所谓"重入",就是同一个线程多次获取锁时,如果此线程已经持有了锁(即创建了zk临时节点),
    事实上将不会再次创建zk的临时节点,而是直接返回.

    因为"重入锁",基于临时节点的特性,因此必须关注client链接重建的问题;粗糙的解决办法,就是每次链接重建(session过期),
    重新实例化lock对象. 
 * @author litianyi
 *
 * 2015年11月3日
 */
public class DistributedLock {
	
	 private InterProcessMutex lock;//重入的,排他的.
	    private Map<Thread,Boolean> lockedThread = new WeakHashMap<Thread,Boolean>();

	    private String lockPath;
	    
	    private String clientName;
	    
	    private ConnectionStateListener stateListener = new StateListener();

	    private RevocationListener<InterProcessMutex> revocationListener;
	    
	    
	    public DistributedLock(CuratorFramework client,String path,String clientName){
	        lockPath = path;
	        this.clientName = clientName;
	        revocationListener = new RevocationListener<InterProcessMutex>() {
	            @Override
	            public void revocationRequested(InterProcessMutex forLock) {
	            	 System.out.println("revocationListener="+forLock.isAcquiredInThisProcess()+"");
	                if(!forLock.isAcquiredInThisProcess()){
	                    return;
	                }
	               
	                try{
	                    forLock.release();
	                }catch(Exception e){
	                    e.printStackTrace();
	                }
	            }
	        };
	        lock = createLock(client);
//	        将锁设为可撤销的. 当别的进程或线程想让你释放锁是Listener会被调用。
	        lock.makeRevocable(revocationListener);
//	        还是强烈推荐你使用ConnectionStateListener处理连接状态的改变。 当连接LOST时你不再拥有锁。
	        client.getConnectionStateListenable().addListener(stateListener);
	    }

	    public boolean lock(){
	        try{
	        	System.out.println("client="+clientName+"lock");
//	        	通过acquire获得锁，并提供超时机制：
	            lock.acquire();
	            lockedThread.put(Thread.currentThread(),Boolean.TRUE);
	        } catch (Exception e){
	            //
	        }
	        return false;
	    }

	    public void unlock(){
	        try{
	        	System.out.println("client="+clientName+"unlock");
	            lock.release();
	        }catch (Exception e){
	            //
	        }
	    }
	    

	    private InterProcessMutex createLock(CuratorFramework client){
	        lock = new InterProcessMutex(client,lockPath);
	        //协同中断,如果其他线程/进程需要此锁中断时,调用此listener.
	        lock.makeRevocable(revocationListener);
	        client.getConnectionStateListenable().addListener(stateListener);
	        return lock;
	    }

	    
	    class StateListener implements ConnectionStateListener{
	        @Override
	        public void stateChanged(CuratorFramework client, ConnectionState newState) {
	            if(Boolean.FALSE.equals(lockedThread.get(Thread.currentThread()))){
	                return;//如果当前lock没有获取锁,则忽略
	            }
	            switch (newState){
	                case LOST:
	                	 System.out.println(newState.toString());
	                    //一旦丢失链接,就意味着zk server端已经删除了锁数据
	                    lockedThread.clear();
	                    lock = createLock(client);//must be rebuild
	                    break;
	                default:
	                    System.out.println(newState.toString());
	            }
	        }
	    }
}
